結合Python與機器學習 — 建構棒球比賽換投模型

113193510林明杉

2025-06-02

Abstract

Introduction

    這一份研究是針對我與組員原本一起研究的題目〈大聯盟投球時鐘政策對投手之影響 - 以2022、2023投手成績為例〉當中第三個研究問題,做更完整且具體的說明。除了對既有結果進行回顧,我們還會進一步整理模型選擇的理路、程式碼實作的架構、重要變數的定義、資料處理的步驟,以及模型訓練時的參數設定與驗證流程。透過這些細節,我們希望讓研究過程透明化,讓其他隊友或後續研究者能夠在相同框架下複製並調整,以因應不同情境的需求。
    一直以來,不管在什麼棒球聯盟,正確的換投時機都是所有教練鍥而不捨想要追求的目標。要預測先發投手當下是否還能穩定投球,不僅要評估他眼下的狀況,還要綜合考量比賽局勢、投手的身體與心理狀態等多種因素。當一個判斷稍有錯誤,就可能導致球隊失去關鍵局面,甚至改變整場比賽的走向。例如,投手投到疲勞階段時,被打者針對性攻擊的風險會大幅提高;而教練在決定換投時,也會將這些實務面的判斷納入考量。因此,我們在模型設計時,會融入這些實務條件,讓分析結果更貼近教練實際決策需求。
    在後續章節中,我們會介紹如何運用版本控制來管理程式碼,並將資料處理與模型實作分為明確的模組,以便團隊協作與維護。此外,為了確保所有撰寫內容的可重現性,我們也會說明程式撰寫時的環境配置,包括所使用的套件版本與必要的硬體需求,使他人能夠快速搭建相同的分析環境。最後,我們將提供足夠的程式註解與文件說明,讓讀者能夠輕鬆理解每段程式碼的功能與目的,進一步提升研究成果的可讀性與可分享性。

Research Question

    本研究將專注於
        1. 如何利用現有數據與模型,準確預測先發投手在單場比賽中的用球數與換投時機?

    我們想知道,透過整理投手的基本資料、逐球時間等資訊,能否建立一套工具,讓教練在比賽中及時判斷這位先發投手能否繼續勝任,以及大概會在哪個時刻被換下。這樣的預測可以輔助現場決策,減少因觀察不及時而造成的風險。

Research Method

    本研究原先使用XGBoost(極限梯度提升)及Logistic Regression(邏輯斯回歸)分類器,但因為前者本質上決策樹的延伸,而後者的假設是自變數與目標之間為線性關係,這點我們無法驗證。再者,我們訓練的資料標籤1(換投)與0(不換投)的樣本數極度不平衡。因此訓練出的模型往往會過於保守或樂觀,造成違反常理的換投時機出現。因此我們採取One-Class SVM(單類支援向量機)來訓練資料及建模。
    有別於傳統的SVM,One-Class SVM屬於非監督式學習,在模型訓練上只需要一類資料(通常是正的觀測值),其目的是要為「正常值的資料」建立邊界,以偵測異常樣本。在我們了訓練資料及當中,有換投的row會被當作正常,反之則為異常。我們希望當模型在偵測先在先發投手該場比賽的用球紀錄,能夠判斷該時刻的投球資訊是否為正常該換投的情況(輸出值為1),以進行換投的預測。

圖一、One-Class SVM示意圖
圖一、One-Class SVM示意圖

Research Data

    以下為本研究訓練資料之特點:
        1. 來自Baseball Savant.
        2. 共有40位投手的投球資料。
        3. 時間從2011~2024年。
        4. 總計686149球,175910個打席。
        5. 資料呈現csv檔。

圖二、訓練資料檔案
圖二、訓練資料檔案


圖三、訓練原始資料內部
圖三、訓練原始資料內部

Feature Engineering

    Feature Engineering(特徵工程)主要是為了篩選出模型預測最有幫助的變數,以提升模型性能與解釋力。
    在原始資料眾多欄位中,我們選取要丟如模型訓練的欄位為pitch_count, release_speed_diff, release_spin_rate_diff, stand, reach_base, last_5以及change。值得注意的是,以上萃取出的欄位幾乎都不是原始資料,而是從原始資料後,經過Python的運算所獲得的「自定義」資料。以下為各個重要欄位之解說:
        1. pitch_count: 該投手該場的累計投球數
        2. release_speed_diff: 該投手該場該球種球速與該投手該場該球種平均球速之差。
        3. release_spin_rate_diff: 該投手該場該球種轉速與該投手該場該球種平轉球速之差。
        4. stand: 該投手該打席遇到打者之打擊型態(左打或又打)
        5. reach_base: 該投手該場累計被上壘次數
        6. last_5:該投手最近五個打席被上壘的次數
        7. 該打席結束後是否有換投手(0表示沒有,1表示有)

    以下為feature engineering的程式碼:

def FE(DF):
    # 移除轉速缺失值
    DF = DF.dropna(subset=['release_spin_rate'])

    # 加入 game_id:每場唯一的場次編號
    DF = DF.sort_values(['game_date', 'player_name']).reset_index(drop = True)
    DF['game_id'] = (
        (DF['game_date'] != DF['game_date'].shift()) | 
        (DF['player_name'] != DF['player_name'].shift())
    ).cumsum() - 1

    # 萃取需要的欄位
    df = DF[['game_id', 'game_date', 'player_name', 'pitch_type', 'release_speed', 'release_spin_rate',
             'events', 'description', 'stand', 'des']].copy()

    hit_events = ['single', 'double', 'triple', 'home_run', 'walk', 'hit_by_pitch']
    df['events'] = df['events'].fillna(df['description'])

    # 每場比賽重新編號 pitch_count
    df['pitch_count'] = df.groupby('game_id').cumcount() + 1

    # 標記是否上壘、打席分組
    df['hit_flag'] = df['events'].isin(hit_events).astype(int)
    df['atbat_id'] = (df['des'] != df['des'].shift()).cumsum()

    # 該場該球種的平均數
    df['avg_speed'] = df.groupby(['game_id', 'pitch_type'])['release_speed'].transform('mean')
    df['avg_spin'] = df.groupby(['game_id', 'pitch_type'])['release_spin_rate'].transform('mean')

    df['release_speed_diff'] = df['release_speed'] - df['avg_speed']
    df['release_spin_rate_diff'] = df['release_spin_rate'] - df['avg_spin']

    # 每打席平均偏差(只保留最後一球)
    grouped = df.groupby(['game_id', 'atbat_id'])
    df['release_speed_diff_mean'] = grouped['release_speed_diff'].transform('mean')
    df['release_spin_rate_diff_mean'] = grouped['release_spin_rate_diff'].transform('mean')
    df['atbat_len'] = grouped['release_speed_diff'].transform('count')
    df = df.loc[grouped.tail(1).index].copy()

    df['release_speed_diff'] = (df['release_speed_diff_mean'] / df['atbat_len']).round(2)
    df['release_spin_rate_diff'] = (df['release_spin_rate_diff_mean'] / df['atbat_len']).round(2)

    # 重排順序
    df = df.sort_values(['game_id', 'pitch_count']).reset_index(drop = True)

    # 修正 reach_base & last_5
    df['reach_base'] = (
        df.groupby('game_id')['hit_flag']
          .transform(lambda x: x.shift().fillna(0).cumsum().astype(int))
    )
    df['last_5'] = (
        df.groupby('game_id')['hit_flag']
          .transform(lambda x: x.shift().rolling(5, min_periods=1).sum().fillna(0).astype(int))
    )

    # 每場最後一球為 change
    last_pitch = df.groupby('game_id')['pitch_count'].transform('max')
    df['change'] = (df['pitch_count'] == last_pitch).astype(int)

    # 最終欄位
    df = df[['game_id', 'player_name', 'des', 'pitch_count',
             'release_speed_diff', 'release_spin_rate_diff', 'stand',
             'reach_base', 'last_5', 'change']]

    return df

    以下為feature engineering後,準備當作訓練資料的結構:

圖四、特徵工程後之訓練資料部分擷取
圖四、特徵工程後之訓練資料部分擷取

Model Training

    進行feature engineering之後,接著就是正式訓練One-Class SVM模型了!在繁雜的程式碼當中,有些值得注意的看點:
        1. 這個訓練方法只會取標記[‘change’] == 1的資料進行訓練。
        2. 本研究會將訓練資料中不同欄位皆進行標準化,使得不同資料的尺度盡可能接近,最小化影響結果。
        3. 我們取0.9為判斷異常資料的閾值,當模型輸出的score < 0.9,表示異常值(不該換投)。
        4. 訓練完畢的模型會分別存為兩個pkl檔。

    以下為model training的程式碼:

def Oneclass_SVM_model(DF, prob_threshold, TT):
    
    global md
    
    features = ['pitch_count', 'release_speed_diff', 'release_spin_rate_diff',
                'reach_base', 'last_5']

    # 只取 change=1 的例子作為訓練資料(表示該換投的情況)
    train_df = DF[DF['change'] == 1].dropna(subset = features).copy()
    X_train = train_df[features]

    # 全資料預測(移除 NaN)
    X_all = DF[features].copy()
    X_all_clean = X_all.dropna()
    all_index = X_all_clean.index

    # 標準化
    scaler = StandardScaler()
    X_train_scaled = scaler.fit_transform(X_train)
    X_all_scaled = scaler.transform(X_all_clean)

    # One-Class SVM 模型
    model = OneClassSVM(kernel = 'rbf', gamma = 'auto', nu = 0.1)
    model.fit(X_train_scaled)

    # decision_function 分數越小越不正常(越不像換投的樣子)
    decision_scores = model.decision_function(X_all_scaled)
    DF['change_score'] = np.nan
    DF.loc[all_index, 'change_score'] = decision_scores

    # 設定閾值標記 should_change(可用分數分布自行調整)
    DF['should_change'] = (DF['change_score'] < np.quantile(decision_scores, 1 - prob_threshold)).astype(int)

    joblib.dump(model, md + TT + ' ' + 'Oneclass_SVM_model.pkl')
    joblib.dump(scaler, md + TT + ' ' + 'Oneclass_SVM_scaler.pkl')

    return DF, model, scaler

    以下為訓練完的模型檔案形式:

圖五、模型檔案

Simulated Game Prediction on Pitching Change

    因為在訓練集中,正確換投的標籤數量甚少,我們研判這個模型的confusion matrix, f1_score等等指標並沒有太大的參考性。因此我決定自行生成一場比賽的投球資料,並交由模型判斷哪個時機點是換投的最佳時機,藉此參考模型的效果及合理性。值得一提的是,上一段落有提到0.9的閾值,我們對這個閾值有制定相關的換投警示規則:
        1. 每場比賽中,每個打席結束若超過閾值0.9,則會給出一個“warning!!!”警告
        2. 若同一場中(滿足以下任一條件):
            2.1 出現第二次連續兩個打席有”warning!!!“的情況。
            2.2 連續面對三個打席後皆出現warning!!!”。
            2.3 最近面對五個打席,有三次“warning!!!”。
            2.4 整場累積達五個打席有”warning!!!“。
         則會給出“change!!!”警告,表示應該要換投手了。

    以下為simulated game prediction on pitching change的程式碼:

def predict(df_new, trained_model, scaler, threshold, output_name, use_weight = True):
    features = ['pitch_count', 'release_speed_diff', 'release_spin_rate_diff', 'reach_base', 'last_5']
    df_new = df_new.copy()

    # 保留原始球速與轉速差
    df_new['original_release_speed_diff'] = df_new['release_speed_diff']
    df_new['original_release_spin_rate_diff'] = df_new['release_spin_rate_diff']

    df_new = df_new.dropna(subset = features)

    # 權重倍率調整(依 pitch_count)
    if use_weight:
        weights = df_new['pitch_count'].apply(
            lambda pc: 0.2 if pc <= 10 else 0.4 if pc <= 20 else 0.6 if pc <= 30 else 0.8 if pc <= 40 else 1.0
        )
        df_new['release_speed_diff'] *= weights
        df_new['release_spin_rate_diff'] *= weights

    # 標準化
    X = df_new[features]
    X_scaled = scaler.transform(X)

    # 預測與標準化成百分比
    decision_score = trained_model.decision_function(X_scaled)
    proba_like = (decision_score - decision_score.min()) / (decision_score.max() - decision_score.min())
    df_new['change_proba'] = ['{:.2f}%'.format(p * 100) for p in proba_like]
    df_new['warning'] = np.where(proba_like > threshold, "warning!!!", "")
    df_new['should_change'] = ""

    for gid, group in df_new.groupby('game_id'):
        group = group.reset_index()
        warnings = group['warning'].values
        change_flags = [""] * len(group)

        warning_count = 0
        double_warning_streaks = 0
        current_streak = 0
        second_double_found = False

        for i in range(len(group)):
            is_warning = warnings[i] == "warning!!!"
            if is_warning:
                current_streak += 1
                warning_count += 1
            else:
                current_streak = 0

            if current_streak == 2 and not second_double_found:
                double_warning_streaks += 1
                if double_warning_streaks == 2:
                    second_double_found = True

            cond0 = current_streak >= 3
            cond1 = second_double_found
            cond2 = np.sum(warnings[max(0, i - 4):i + 1] == "warning!!!") >= 3
            cond3 = warning_count >= 5

            if cond0 or cond1 or cond2 or cond3:
                change_flags[i] = "change!!!"

        df_new.loc[group['index'], 'should_change'] = change_flags

    # 還原原始欄位
    df_new['release_speed_diff'] = df_new['original_release_speed_diff']
    df_new['release_spin_rate_diff'] = df_new['original_release_spin_rate_diff']
    df_new.drop(columns=['original_release_speed_diff', 'original_release_spin_rate_diff'], inplace = True)

    df_new.to_csv(output_name, index = False, encoding='utf-8-sig')
    print(f"\n結果已儲存為:{output_name}")
    return df_new

    以下為模擬比賽原始數據的待測試檔案部分內容:

圖六、模擬比賽原始數據的待測試檔案部分內容

Research Result

    經過模型的訓練及測試資料的生成,以下為套用換投警示系統模型在模擬比賽的結果:

圖七、模擬比賽換投結果

    從模擬結果,可以看到以下幾個值得注意的現象:
        1. change_proba(模型輸出的換投機率)大致上隨球數逐漸升高
        2. “warning!!!”及”change!!!“出現的時間皆為合理範圍
        3. 警示/換投標記邏輯符合一般教練直覺,當last_5跟reach_base的情況沒有明顯好轉時,會適時觸發警報。

    因此,我們認為上述幾點已足以證明模型訓練的成效。模型能夠在關鍵球數階段準確地發出”warning!!!“並於連續多次警示後自動標示”change!!!“,反映出特徵選取與參數設定的有效性,並與教練實務判斷高度一致。未來若在更多真實比賽資料中進一步驗證,相信此模型能為場邊決策提供更可靠的參考。

Conclusion and Discussion

    綜合這整篇來看,我們可以得出以下結論:
        1. 模型成效與實務對應:One-Class SVM 透過僅以「真實換投狀態」作為正例訓練,就能在模擬情境中有效區分「應繼續投球」與「應該換投」兩種狀態。相較於傳統二分類在樣本極度不平衡時容易偏向大類別(多數為不換投),我們的「單類學習+連續警示邏輯」設計更能抓住投手表現惡化的「異常模式」。
        2. 特徵選取與權重機制的影響:我們挑選的指標在模擬結果中確實呈現與換投機率上升同步的趨勢,證明它們具有代表投手「狀態下滑」的資訊量。
        3. 警示與換投邏輯的合理性:採用多個「連續警示」與「累積警示」的規則,可避免單一球次出現波動(例如偶爾球速偏差或突然失誤)就立刻判定換投,有助於降低假警示(false alarm)的機率。
        4. 限制與後續改進(模擬資料vs真實資料):模擬比賽資料可用來驗證模型邏輯的合理性,但畢竟不同於實際大聯盟賽事中的現場干擾(如教練主觀判斷、球隊戰術調度、即時計分結果等)。
        5. 特徵維度不足:本研究僅使用部分「投球相關」衍生指標,尚未納入「球隊整體局勢」(例如當前比分差距、球隊板凳深度、接手牛棚累計投球量)、「投手生理狀態」或「心理指標」(如心率、疲勞指數、球速變化率)等。這些資訊若能與現有特徵結合,將有助於捕捉更多影響換投決策的因素,進一步提升預測精準度。
        6. 未來應用與展望:若能進一步在真實大聯盟賽事資料上回測並與實際換投時機做比較,即可更全面地驗證模型的精準度與穩定性。同時透過引入額外變數(如生理指標、戰術面資料)或比較其他異常偵測算法,便能持續優化系統效能。最終目標是打造一套「可在場邊即時運行、易於解譯且可靠度高」的換投預測工具,協助教練團做出更客觀的決策,降低傳統經驗判斷所帶來的不確定性,並為球隊贏球帶來實質助益。